package com.etc;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

public class WordStepOne extends Mapper<LongWritable, Text, Text, IntWritable> {

    public static class WordStepMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            FileSplit split = (FileSplit) context.getInputSplit();
            String fileName = split.getPath().getName();
            String line = value.toString();
            String[] sipt = line.split(" ");
            for (String s : sipt) {
                context.write(new Text(s +"-"+ fileName), new IntWritable(1));
            }
        }
    }

    public static class WordReduce extends Reducer<Text, IntWritable, Text, IntWritable> {

        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int count = 0;
            for (IntWritable value : values) {
                count += value.get();
            }
            context.write(new Text(key), new IntWritable(count));
        }
    }

}
